package com.hxq.demo6;

import cn.hutool.core.collection.ListUtil;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.acl.*;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.resource.PatternType;
import org.apache.kafka.common.resource.ResourcePattern;
import org.apache.kafka.common.resource.ResourcePatternFilter;
import org.apache.kafka.common.resource.ResourceType;

import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * @author hxq
 * @date 2022/4/27 10:45
 */
public class ACLDemo {
    public static void main(String[] args) throws ExecutionException, InterruptedException, TimeoutException {

        //配置
        Properties properties = new Properties();
        properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,"212.129.155.35:9092,212.129.155.35:9093");
        //设置SASL连接
        properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
        properties.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-512");
        //方式一
        properties.put("sasl.jaas.config","org.apache.kafka.common.security.scram.ScramLoginModule required username=\"admin\" password=\"admin\";");

        //创建
        AdminClient adminClient = AdminClient.create(properties);

        //创建topic
//        createTopic(adminClient,"delay-minutes-1",1,(short) 1);
        //topic列表
//        Set<String> topics = topicList(adminClient);
//        System.out.println(topics);

        //获取某个主题的acl
//        Collection<AclBinding> first = describeAclsByTopicName(adminClient, "second");
//        System.out.println(first);
        //为writer用户分配first主题的写权限
//        createAclsforTopic(adminClient,"delay-minutes-1","writer",AclOperation.WRITE);

        //取消writer1用户对senond主题的写权限
//        denyAclsforTopic(adminClient,"second","writer1",AclOperation.WRITE);

        //删除writer1用户对secon主题的acl
//        delAclForTopic(adminClient,"first","writer1",AclOperation.WRITE,AclPermissionType.ALLOW);

        //为reader用户分配first主题的读权限
//        createAclsforTopic(adminClient,"delay-minutes-1","reader",AclOperation.READ);
        //为reader用户分配消费者组 test1的权限
//        createAclsforGroup(adminClient,"test","guiyun",AclOperation.ALL);

        //为reader用户删除消费者组 test1的权限
//        Collection<AclBinding> res = deleteAclsForGroup(adminClient, "reader", "test1", AclOperation.READ);
//        System.out.println(res);

        //创建用户
//        createUser(adminClient,"guiyun","123456");

        //获取所有用户信息
//        Map<String, UserScramCredentialsDescription> userCredentialsInfo = getUserCredentialsInfo(adminClient);
//        System.out.println(userCredentialsInfo);

        //修改用户 修改 writer1 的密码
//        updateUser(adminClient,"writer1","123456.");

        //删除writer1用户
//        deleteUser(adminClient,"guiyun");

        //查看topic详情
        TopicDescription first = describeTopics(adminClient, "second");
        System.out.println(first);


    }

    //删除主题的acl
    private static Collection<AclBinding> delAclForTopic(AdminClient adminClient,String topicName,String username,AclOperation auth,AclPermissionType permissionType) throws ExecutionException, InterruptedException {
        ResourcePatternFilter resourcePatternFilter = new ResourcePatternFilter(ResourceType.TOPIC, topicName, PatternType.LITERAL);//资源信息
        AccessControlEntryFilter accessControlEntryFilter = new AccessControlEntryFilter("User:" + username, "*", auth, permissionType);//用户信息
        AclBindingFilter filter = new AclBindingFilter(resourcePatternFilter, accessControlEntryFilter);
        Collection<AclBinding> aclBindings = adminClient.deleteAcls(ListUtil.of(filter)).all().get();
        return aclBindings;
    }

    //查看topic详情
    private static TopicDescription describeTopics(AdminClient adminClient,String topicName) throws ExecutionException, InterruptedException {
        Map<String, KafkaFuture<TopicDescription>> map = adminClient.describeTopics(ListUtil.of(topicName)).topicNameValues();
        TopicDescription topicDescription = map.get(topicName).get();
        return topicDescription;
    }

    //删除用户
    private static void deleteUser(AdminClient adminClient,String username) throws ExecutionException, InterruptedException {

        UserScramCredentialDeletion credentialDeletion = new UserScramCredentialDeletion(username, ScramMechanism.SCRAM_SHA_512);
        adminClient.alterUserScramCredentials(ListUtil.of(credentialDeletion)).all().get();
    }

    //修改用户
    private static void updateUser(AdminClient adminClient,String username, String password) throws ExecutionException, InterruptedException {
        ScramCredentialInfo scramCredentialInfo = new ScramCredentialInfo(ScramMechanism.SCRAM_SHA_512, 8192);
        UserScramCredentialUpsertion create = new UserScramCredentialUpsertion(username, scramCredentialInfo, password);
        adminClient.alterUserScramCredentials(ListUtil.of(create)).all().get();
    }


    //获取所有用户信息
    private static Map<String, UserScramCredentialsDescription> getUserCredentialsInfo(AdminClient adminClient) throws ExecutionException, InterruptedException, TimeoutException {

        DescribeUserScramCredentialsResult credentials = adminClient.describeUserScramCredentials();
        Map<String, UserScramCredentialsDescription> map = credentials.all().get();
        return map;
    }

    //创建用户
    private static void createUser(AdminClient adminClient,String username,String password) throws ExecutionException, InterruptedException {
        ScramCredentialInfo scramCredentialInfo = new ScramCredentialInfo(ScramMechanism.SCRAM_SHA_512, 8192);
        UserScramCredentialUpsertion create = new UserScramCredentialUpsertion(username, scramCredentialInfo, password);
        adminClient.alterUserScramCredentials(ListUtil.of(create)).all().get();
    }

    //为某用户取消某个消费者组的权限
    private static Collection<AclBinding> deleteAclsForGroup(AdminClient adminClient,String userName,String groupName,AclOperation auth)throws ExecutionException, InterruptedException{
        ResourcePatternFilter groupInfo = new ResourcePatternFilter(ResourceType.GROUP,groupName,PatternType.LITERAL);//组信息
        AccessControlEntryFilter userInfo = new AccessControlEntryFilter("User:" + userName, "*", auth, AclPermissionType.ALLOW);//用户信息
        AclBindingFilter filter = new AclBindingFilter(groupInfo, userInfo);
        Collection<AclBinding> res = adminClient.deleteAcls(ListUtil.of(filter)).all().get();
        return res;
    }


    //为某用户分配消费者组的权限
    private static void createAclsforGroup(AdminClient adminClient,String groupName,String userName,AclOperation auth) throws ExecutionException, InterruptedException {

        ResourcePattern topicInfo = new ResourcePattern(ResourceType.GROUP, groupName, PatternType.LITERAL);//定义消费者组信息
        AccessControlEntry userInfo = new AccessControlEntry("User:"+userName, "*", auth, AclPermissionType.ALLOW);//定义用户信息
        AclBinding aclBinding = new AclBinding(topicInfo, userInfo);
        adminClient.createAcls(ListUtil.of(aclBinding)).all().get();
    }

    //对某用户分配某个topic的某个权限
    private static void createAclsforTopic(AdminClient adminClient,String topicName,String userName,AclOperation auth) throws ExecutionException, InterruptedException {

        ResourcePattern topicInfo = new ResourcePattern(ResourceType.TOPIC, topicName, PatternType.LITERAL);//定义topic信息
        AccessControlEntry userInfo = new AccessControlEntry("User:"+userName, "*", auth, AclPermissionType.ALLOW);//定义用户信息
        AclBinding aclBinding = new AclBinding(topicInfo, userInfo);
        adminClient.createAcls(ListUtil.of(aclBinding)).all().get();
    }

    //取消某用户对topic的权限
    private static void denyAclsforTopic(AdminClient adminClient,String topicName,String userName,AclOperation auth) throws ExecutionException, InterruptedException {

        ResourcePattern topicInfo = new ResourcePattern(ResourceType.TOPIC, topicName, PatternType.LITERAL);//定义topic信息
        AccessControlEntry userInfo = new AccessControlEntry("User:"+userName, "*", auth, AclPermissionType.DENY);//定义用户信息
        AclBinding aclBinding = new AclBinding(topicInfo, userInfo);
        adminClient.createAcls(ListUtil.of(aclBinding)).all().get();
    }

    //列出某个主题的acl
    private static Collection<AclBinding> describeAclsByTopicName(AdminClient adminClient,String topicName) throws ExecutionException, InterruptedException {
        AclBindingFilter filter = new AclBindingFilter(new ResourcePatternFilter(ResourceType.TOPIC, topicName, PatternType.LITERAL), new AccessControlEntryFilter(null, null, AclOperation.ANY, AclPermissionType.ANY));
        DescribeAclsResult describeAcls = adminClient.describeAcls(filter);
        Collection<AclBinding> aclBindingCollection = describeAcls.values().get();
        return aclBindingCollection;
    }

    private static void createTopic(AdminClient adminClient,String name,Integer partition,Short replication){

        CreateTopicsResult result = adminClient.createTopics(Arrays.asList(new NewTopic(name, partition, (short) replication)));

        try {
            result.all().get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }

    private static Set<String> topicList(AdminClient adminClient){
        ListTopicsResult result = adminClient.listTopics();
        try {
            return result.names().get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        return new HashSet<>();
    }
}
